package batch;

import org.apache.flink.api.common.operators.Order;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;

import java.util.ArrayList;

public class operator_first_N {
    public static void main(String[] args) throws Exception {
    //    创建运行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    //    获取数据
        ArrayList<Tuple2<Integer,String>> data1 = new ArrayList<Tuple2<Integer,String>>();
        data1.add(new Tuple2<Integer, String>(1,"bing"));
        data1.add(new Tuple2<Integer, String>(2,"shui"));
        data1.add(new Tuple2<Integer, String>(3,"fan"));
        data1.add(new Tuple2<Integer, String>(4,"hong"));
        data1.add(new Tuple2<Integer, String>(2,"yang"));
        data1.add(new Tuple2<Integer, String>(3,"hong"));




        DataSet<Tuple2<Integer,String>> source =  env.fromCollection(data1);


       //source.first(3).print();
    //   根据第一列进行分组，然后第二列组内降序排序，取每组第一个
        source.sortPartition(0, Order.ASCENDING).groupBy(0).sortGroup(1, Order.DESCENDING).first(1).print();

        System.out.println("------------");
        //    第一列升序，第二列降序
        source.sortPartition(0,Order.ASCENDING).sortPartition(1,Order.DESCENDING).print();

    }
}
